package com.minelx.datamicroscope.core.query;

import com.alibaba.fastjson.JSONObject;
import com.minelx.datamicroscope.core.display.DisplayTemplate;
import com.minelx.datamicroscope.core.display.DisplayTemplates;
import com.minelx.datamicroscope.core.display.Merger;
import com.minelx.datamicroscope.core.display.TemplateProcessedMerger;
import com.minelx.hellopandas.core.DataFrame;
import com.minelx.hellopandas.core.PandasChief;
import com.minelx.hellopandas.core.PandasMethodProvider;
import io.gitee.minelx.filesystem.provider.FS;
import io.gitee.minelx.filesystem.provider.FSRegistry;
import io.gitee.minelx.filesystem.values.SchemeAndAuthority;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static io.gitee.minelx.filesystem.init.OpenPolicy.EXISTS;

public class Query {

	private final List<DisplayTemplate> projection;

	private final DisplayTemplates displayTemplates;

	private final QueryPaths queryPaths;

	private final Map<String, String> howMapping;

	Query(List<DisplayTemplate> projection,
		  DisplayTemplates displayTemplates,
		  QueryPaths queryPaths,
		  Map<String, String> howMapping) {
		this.projection = projection;
		this.displayTemplates = displayTemplates;
		this.queryPaths = queryPaths;
		this.howMapping = howMapping;
	}

    public void queryTo(String destinationURI) {
		// do query
		DataFrame queriedDataFrame = queryPaths.query(
				destinationURI,
				new TemplateProcessedMerger(Merger.join(howMapping), displayTemplates));
		// do projection
		//noinspection ResultOfMethodCallIgnored @ effective method DisplayTemplate.process
		projection.stream().reduce(queriedDataFrame,
				(dataFrame, displayTemplate) -> displayTemplate.process(dataFrame),
				(selectThe, lastOne) -> lastOne);
//		projection.forEach(displayTemplate -> displayTemplate.process(queriedDataFrame));
	}

	public static Query create(ResourcesFactory resourcesFactory, JSONObject data) {
		return new QueryFactory(resourcesFactory, data).createQuery();
	}

	public static Query localFSAndOnlineIndexWithPandasCalculationSupport(JSONObject data) {
		return create(ResourcesFactory.localFSWithGeneralIndex(), data);
	}

	public static FSRegistry onlyLocalFSRegistry() {
		Map<SchemeAndAuthority, FS> result = new HashMap<>();
		result.put(SchemeAndAuthority.from("file://"), FS.LOCAL_FS);
		return FSRegistry.create(result);
	}

	public static PandasMethodProvider localPandas() {
		return PandasMethodProvider.create(PandasChief.at("127.0.0.1", 5000));
	}

	static String toText(DataFrame dataFrame) {
		try {
			return onlyLocalFSRegistry().file(dataFrame.description().uri(), EXISTS)
					.read(StandardCharsets.UTF_8);
		} catch (IOException e) {
			throw new RuntimeException("error while reading content.", e);
		}
	}
}
